package com.dji.pa.cluster.db.dataSouce;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.aspectj.lang.ProceedingJoinPoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.interceptor.NameMatchTransactionAttributeSource;
import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.ReflectionUtils;

public class DataSourceProcessor implements BeanPostProcessor,InitializingBean{
    private final Logger log=LoggerFactory.getLogger(DataSourceProcessor.class);
	private final Map<String, Boolean> readMethodMap=new HashMap<String, Boolean>();
	private AbstractClusterDataSource clusterDataSource;
	private boolean forceChoiceReadWhenWrite=false;
	private boolean enableAsWriteWhenWriteDown=false;
	private boolean enableAsReadWhenAllReadDown=false;
	private int failedTimeToTryAgain=2;
	private ServerWatcher watcher;
	private Lock lock;
	private RandomLoadBalance balance=null;

	/**
	 * 创建监控
	 */
	public void afterPropertiesSet() throws Exception{
		balance=new RandomLoadBalance();
		watcher=new ServerWatcher(clusterDataSource,lock=new ReentrantLock());
		Thread watcherThread=new Thread(watcher);
		watcherThread.setDaemon(true);
		watcherThread.start();
	}

	/**
	 * 2
	 *
	 * @param pjp
	 * @return
	 * @throws Throwable
	 */
	public Object determineReadOrWriteDB(ProceedingJoinPoint pjp) throws Throwable{
		String methodName=pjp.getSignature().getName();
		int time=0;
		try{
			if(isChoiceReadDB(methodName)){
				ReadWriteDecision.markRead();// 读库进行失败重试机制
				while(time<=failedTimeToTryAgain){
					try{
						return pjp.proceed();
					}catch(Throwable e){
						if(CommonUtils.isCommunicationsException(e)){
							time++;
							switchDBIfReadError(pjp,e);
						}else{
							break;
						}
					}
				}
			}else{
				ReadWriteDecision.markWrite();
				while(time<=failedTimeToTryAgain){
					try{
						return pjp.proceed();
					}catch(Throwable e){
						if(CommonUtils.isCommunicationsException(e)){
							time++;
							switchDBIfWriteError(pjp,e);
						}else{
							break;
						}
					}
				}
			}
		}finally{
			ReadWriteDecision.reset();// 此处顺序不能颠倒
		}
		return null;
	}

	/**
	 * 获取最匹配的服务器
	 *
	 * @return
	 */
	public Server getMatchServer(){
		Server result=null;
		if(ReadWriteDecision.isChoiceWrite()){
			// TODO 如果有自己的写分片算法，这里要改动
			result=clusterDataSource.getWriteDBList().get(0);
			log.debug("选择器：写库"+result.getIpAndPort());
		}else if(ReadWriteDecision.isChoiceRead()){
			List<Server> readDBList=clusterDataSource.getReadDBList();
			log.debug(readDBList.toString());
			result=balance.doSelect(readDBList);
			log.debug("选择器：读库"+result.getIpAndPort());
		}else{// 未知
			result=clusterDataSource.getWriteDBList().get(0);
			log.debug("选择器：写库,不符合标准的方法"+result.getIpAndPort());
		}
		clusterDataSource.getHolder().set(result);
		return result;
	}

	/**
	 * 2.1
	 *
	 * @param methodName
	 * @return
	 */
	private boolean isChoiceReadDB(String methodName){
		String bestNameMatch=null;
		for(String mappedName:this.readMethodMap.keySet()){
			if(PatternMatchUtils.simpleMatch(mappedName,methodName)){
				bestNameMatch=mappedName;
				break;
			}
		}
		Boolean isForceChoiceRead=readMethodMap.get(bestNameMatch);
		// 表示强制选择 读 库
		if(isForceChoiceRead==Boolean.TRUE){
			return true;
		}
		// 如果之前选择了写库 现在还选择 写库
		if(ReadWriteDecision.isChoiceWrite()){
			return false;
		}
		// 表示应该选择读库
		if(isForceChoiceRead!=null){
			return true;
		}
		// 默认选择 写库
		return false;
	}

	/**
	 * 1
	 */
	@SuppressWarnings("unchecked")
	public Object postProcessAfterInitialization(Object bean,String beanName) throws BeansException{
		if(!(bean instanceof NameMatchTransactionAttributeSource)){
			return bean;
		}
		try{
			NameMatchTransactionAttributeSource nmtas=(NameMatchTransactionAttributeSource)bean;
			Field nameMapField=ReflectionUtils.findField(NameMatchTransactionAttributeSource.class,"nameMap");
			nameMapField.setAccessible(true);
			Map<String, TransactionAttribute> nameMap=(Map<String, TransactionAttribute>)nameMapField.get(nmtas);
			for(Entry<String, TransactionAttribute> entry:nameMap.entrySet()){
				RuleBasedTransactionAttribute attr=(RuleBasedTransactionAttribute)entry.getValue();
				if(!attr.isReadOnly()){// 仅对read-only的处理
					continue;
				}
				Boolean isForceChoiceRead=Boolean.FALSE;
				if(forceChoiceReadWhenWrite){
					attr.setPropagationBehavior(Propagation.NOT_SUPPORTED.value());
					isForceChoiceRead=Boolean.TRUE;
				}else{
					attr.setPropagationBehavior(Propagation.SUPPORTS.value());
				}
				log.debug(entry.getKey()+"   ,  "+isForceChoiceRead);
				readMethodMap.put(entry.getKey(),isForceChoiceRead);
			}
		}catch(Exception e){
			throw new RuntimeException("选择器：处理读写事务异常",e);
		}
		return bean;
	}

	public Object postProcessBeforeInitialization(Object bean,String beanName) throws BeansException{
		return bean;
	}

	public void setClusterDataSource(AbstractClusterDataSource clusterDataSource){
		this.clusterDataSource=clusterDataSource;
	}

	public void setEnableAsReadWhenAllReadDown(boolean enableAsReadWhenAllReadDown){
		this.enableAsReadWhenAllReadDown=enableAsReadWhenAllReadDown;
	}

	public void setEnableAsWriteWhenWriteDown(boolean enableAsWriteWhenWriteDown){
		this.enableAsWriteWhenWriteDown=enableAsWriteWhenWriteDown;
	}

	public void setFailedTimeToTryAgain(int failedTimeToTryAgain){
		this.failedTimeToTryAgain=failedTimeToTryAgain;
	}

	public void setForceChoiceReadWhenWrite(boolean forceChoiceReadWhenWrite){
		this.forceChoiceReadWhenWrite=forceChoiceReadWhenWrite;
	}

	/**
	 * 2.2 此处存在多线程问题，来异常的时候会大量的请求进入此方法
	 *
	 * @param pjp
	 * @param class1
	 * @throws InterruptedException
	 */
	private void switchDBIfReadError(ProceedingJoinPoint pjp,Throwable error) throws InterruptedException{
		Server failServer=clusterDataSource.getHolder().get();
		log.warn("选择器：数据库down机"+failServer.getIpAndPort());
		if(!watcher.getFailServerList().contains(failServer) && lock.tryLock(1,TimeUnit.SECONDS)){
			log.debug("选择器：11111111111111111111111   获取锁");
			try{
				String failIpAndPort=failServer.getIpAndPort();
				List<Server> readServerList=clusterDataSource.getReadDBList();
				Iterator<Server> iterator=readServerList.iterator();
				Server server=null;
				while(iterator.hasNext()){
					server=iterator.next();
					if(server.getIpAndPort().equals(failIpAndPort)){
						iterator.remove();
						log.warn("选择器：数据库down机,移除读机器"+failIpAndPort);
						if(readServerList.size()==0&&enableAsReadWhenAllReadDown){
							if(clusterDataSource.getWriteDBList().size()>0){
								log.warn("选择器：警告,读服务器列表为空,将尝试在写服务器进行查询"
										+clusterDataSource.getWriteDBList().get(0).getServerType());
								readServerList.add(clusterDataSource.getWriteDBList().get(0));// 读写共用：读库全部挂掉了就去写库查,这在高并发的时候是坚决不允许的，但是业务量少的时候可以接受
							}else{
								log.debug("选择器：警告,尝试切换写服务器做读操作...写服务器不可用");
							}
						}
						watcher.getFailServerList().add(failServer);
						break;
					}
				}
			}catch(Exception e){
				// 获取锁超过一秒钟的异常不处理
			}finally{
				lock.unlock();
				log.debug("选择器：1111111111111111   释放锁");
				synchronized(lock){
					lock.notify();
					log.debug("选择器：获取读服务器失败-通知监视器运行");
				}
			}
		}

	}

	/**
	 * 写数据异常时候切库
	 *
	 * @param pjp
	 * @param error
	 * @throws InterruptedException
	 */
	private void switchDBIfWriteError(ProceedingJoinPoint pjp,Throwable error) throws InterruptedException{
		log.warn("选择器：重试新的写服务器);");
		if(CommonUtils.isCommunicationsException(error)){
			Server failServer=clusterDataSource.getHolder().get();
			if(!watcher.getFailServerList().contains(failServer)&& lock.tryLock(1,TimeUnit.SECONDS)){
				log.debug("选择器：222222222222222   获取锁");
				try{
					List<Server> readServerList=clusterDataSource.getReadDBList();
					List<Server> writeServerList=clusterDataSource.getWriteDBList();
					if(writeServerList.remove(failServer)){// 多线程条件下，多线程发现的异常是同一个数据源，只要处理一个就行了
						log.warn("选择器：数据库down机,移除写服务器"+failServer.getIpAndPort());
						if(enableAsWriteWhenWriteDown){// 如果允许读服务器作为写服务器
							if(readServerList.size()==1){
								writeServerList.add(readServerList.get(0));
							}else if(readServerList.size()>1){
								// 在拥有多台读服务器的情况下，第一台读服务器切换为写服务器，其他为写
								Server server=readServerList.get(0);
								writeServerList.add(server);
								readServerList.remove(server);
								log.warn("选择器：由于写服务器"+failServer.getIpAndPort()+"down机，读服务器"+server.getIpAndPort()
										+"切换为主写服务器");
							}else{
								log.warn("选择器：警告,当前已经没有可用的数据库服务器，读写操作均不可用");
							}
							failServer.getIpAndPort();
						}
						watcher.getFailServerList().add(failServer);
					}
				}catch(Exception e){
					 log.warn(e.getMessage(),e);
				}finally{
					lock.unlock();
					log.debug("选择器：222222222222222   释放锁");
					synchronized(lock){
						lock.notify();
						log.debug("选择器：获取写服务器失败-通知监视器运行");
					}
				}
			}
		}
	}
}